package redis.task;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 一个基于Redis中介的异步任务管理类。管理人员向指定设备下发指令时，设备可能连接了其它服务器，同时等待设备返回数据需要较长时间。
 * DeferredResultTaskManager通过Redis Channel发送任务数据，并通过Redis Channel接收任务执行结果。
 * 实现思路：
 * 1. 收到任务指令后，新建空的任务结果实例R，并存入任务结果Map resultMap中
 * 2. 从redis中获取设备session，得到设备连接的server S
 * 3. 向server S监听的任务Channel发送任务数据，新开线程T等待结果R的通知，返回ChannelFuture实例F
 * 4. 本地服务器在redis的任务结果Channel监听到任务结果，调用DeferredResultTaskManager.receivedResult
 * 5. DeferredResultTaskManager从resultMap找到任务结果R，并将任务结果数据复制到R中，然后通知R
 * 6. 线程T收到通知后，向F写入最后结果，线程T结束
 */
public class DeferredResultTaskManager {
    private ObjectMapper mapper;
    private String serverId; // 本机实例ID
    private LocalTaskExecutor executor;
    private StringRedisTemplate redisTemplate;
    private ConcurrentHashMap<String, DeferredResultTaskResult> resultMap;

    public DeferredResultTaskManager(StringRedisTemplate redisTemplate, LocalTaskExecutor executor, String serverId) {
        this.redisTemplate = redisTemplate;
        this.serverId = serverId;
        this.executor = executor;
        this.mapper = new ObjectMapper();
        this.resultMap = new ConcurrentHashMap<>();
    }

    /**
     * 执行任务
     * @param task
     */
    public <T> void executeTask(DeferredResultTask<T> task) {
        if (task.getServerId().equals(serverId)) {
            executor.execute(task);
        } else {
            sendTaskToServer(task);
        }
    }

    /**
     * 执行任务并异步等待结果
     * @param task
     * @param <T>
     * @param <R>
     * @return
     */
    public <T, R> CompletableFuture<DeferredResultTaskResult<R>> executeAndGetResult(DeferredResultTask<T> task) {
        executeTask(task);
        CompletableFuture<DeferredResultTaskResult<R>> future = new CompletableFuture<>();
        DeferredResultTaskResult<R> result = new DeferredResultTaskResult<>(task);
        resultMap.put(task.getServerId() + task.getTaskId(), result);
        new Thread(() -> {
            synchronized (task) {
                try {
                    task.wait(task.getTimeout() * 1000);
                } catch (InterruptedException e) {
                    resultMap.remove(result.getTaskId());
                    result.setResult(DeferredResultTaskResult.ResultState.timeout);
                    future.complete(result);
                    return;
                }
                future.complete(result);
            }
        }).start();
        return future;
    }

    /**
     * 向其它主机发送任务。
     * @param task 由其它主机处理的任务
     */
    private void sendTaskToServer(DeferredResultTask task) {
        try {
            redisTemplate.convertAndSend(getTaskChannel(task.getServerId()), mapper.writeValueAsString(task));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 收到任务结果后，更新任务结果并通知等待线程
     * @param result
     */
    @SuppressWarnings("unchecked")
    public void receivedResult(DeferredResultTaskResult result) {
        DeferredResultTaskResult local = resultMap.remove(result.getTaskId());
        if (local != null) {
            local.copy(result);
            local.notify();
        }
    }

    /**
     * 本机接收任务的Channel
     * @return
     */
    public String getTaskChannel() {
        return "task@" + serverId;
    }

    /**
     * 本机接收任务结果的Channel
     * @return
     */
    public String getResultChannel() {
        return "result@" + serverId;
    }

    /**
     * 接收任务的Channel
     * @return
     */
    public static String getTaskChannel(String serverId) {
        return "task@" + serverId;
    }

    /**
     * 接收任务结果的Channel
     * @param serverId
     * @return
     */
    public static String getResultChannel(String serverId) {
        return "result@" + serverId;
    }
}
